package com.eagle.epay.controller;


import com.eagle.epay.entity.Alarm;
import com.eagle.epay.entity.User;
import com.eagle.epay.entity.api.MonitorApi;
import com.eagle.epay.lock.DistributedLocker;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.logging.HttpLoggingInterceptor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.web.bind.annotation.*;
import retrofit2.Call;
import retrofit2.Response;
import retrofit2.Retrofit;
import retrofit2.converter.gson.GsonConverterFactory;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 *  前端控制器
 * </p>
 *
 * @author eagle
 * @since 2019-12-16
 */
//@Controller
@RestController
//@RequestMapping("/redis")
@Slf4j
public class RedisController {
	@Resource
	private RedisTemplate<String, Object> redisTemplate;

	@Resource
	private RedisTemplate<String, String> redisStringTemplate;

	@Autowired
	private DistributedLocker distributedLocker;

	private int minFrequency = 1;
	private int countFrequency = 2;

	private Retrofit retrofit = new Retrofit.Builder()
			.baseUrl("http://localhost:8088")
			.addConverterFactory(GsonConverterFactory.create())
			.client(new OkHttpClient.Builder()
					.addInterceptor((new HttpLoggingInterceptor()
							.setLevel(HttpLoggingInterceptor.Level.BODY)))
					.build()
			)
			.build();
	@GetMapping("hello")
	public String hello(){
		return "Hello Spring Boot 2";
	}



	@GetMapping("testRetrofilt")
	@ResponseBody
	public String testRetrofilt() throws Exception{
		Alarm alarm = new Alarm();
		alarm.setApp("app");
		alarm.setContent("content");
		Call<okhttp3.ResponseBody> call = retrofit.create(MonitorApi.class).alarm(alarm);
		Response<okhttp3.ResponseBody> response = null;
		try{
			response = call.execute();
		}catch (IOException e) {
			e.printStackTrace();
		}
		return  (null == response) ? "no data" : response.body().string();
	}

	@PostMapping("/monitor/alarm")
	@ResponseBody
	public String callMe(@RequestBody Alarm alarm){
		System.out.println( alarm.getApp()+"-"+alarm.getContent());
		StringBuffer sb = new StringBuffer();
		sb.append(alarm.getApp()).append("-").append(alarm.getContent());
		return sb.toString();
	}


	@GetMapping("/getKey")
	@ResponseBody
	public Object  testRedis(String key){
		Object result = redisTemplate.opsForValue().get(key);
		return result;
	}

	@GetMapping("/setKey")
	@ResponseBody
	public Object setRedis(String key){
		User user = new User();
		user.setId(1);
		user.setUsername("eagle");
		user.setPhone("12345678");
		redisTemplate.opsForValue().set(key,user);
		return "Done";
	}

	@GetMapping("/testRedission")
	@ResponseBody
	public Object testRedission(String key){
		String lockey = "eagle-lock";
		boolean isGetLock = distributedLocker.tryLock(lockey, TimeUnit.SECONDS, 0L, 60L); // 尝试获取锁，自己获得锁后一直不解锁则60s秒后自动解锁
		if(isGetLock){
			try {
				System.out.println("Sleep 3s...");
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}finally {
				// 释放锁
				System.out.println("release lock...");
				distributedLocker.unlock(lockey);
			}
		}
		return "Done";
	}

	@GetMapping("/counter")
	@ResponseBody
	public String counter(String key){
		ZSetOperations<String, String> opt = redisStringTemplate.opsForZSet();
		opt.incrementScore("errorCounter", key, 1);//向集合book增加键

		Double count = opt.score("errorCounter", key);
		if(count >5 ){
			System.out.println("触发清0...");
			opt.remove("errorCounter",key);
		}
		return key+"="+opt.score("errorCounter", key);
	}

	@GetMapping("/zset")
	@ResponseBody
	public Object zset(String key){
		flushdb();
		ZSetOperations<String, String> vo = redisStringTemplate.opsForZSet();
		vo.add("book", "a", 1d);//向集合book增加键
		vo.add("book", "b", 2d);
		Set<ZSetOperations.TypedTuple<String>> tuples = new HashSet<>();
		ZSetOperations.TypedTuple<String> tuple0 = new DefaultTypedTuple<String>("c", 3d);
		tuples.add(tuple0);
		ZSetOperations.TypedTuple<String> tuple1 = new DefaultTypedTuple<String>("d", 4d);
		tuples.add(tuple1);
		ZSetOperations.TypedTuple<String> tuple2 = new DefaultTypedTuple<String>("e", 5d);
		tuples.add(tuple2);

		vo.add("book", tuples);

		vo.incrementScore("book", "e", 1d);//给e加分


		vo.add("book", "f", 7);
		vo.add("book", "g", 8);
		vo.add("book", "h", 9);

//        vo.remove("book", "g","h");
		vo.removeRange("book", 7, 8);
		vo.removeRangeByScore("book", 8, 9);

		return "Done";
	}


	public void flushdb(){
		redisStringTemplate.execute(new RedisCallback<Object>() {
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				connection.flushDb();
				return "ok";
			}
		});
	}

	@GetMapping("/zget")
	@ResponseBody
	public String zget(){
		ZSetOperations<String, String> vo = redisStringTemplate.opsForZSet();
		out(vo.score("book", "e")); //分数
		out(vo.score("book", "f")); //分数
		out(vo.rank("book", "e")); //键值排名（从0开始）
		out(vo.reverseRank("book", "e")); //分值排名
		out(vo.zCard("book")); //集合长度
		out(vo.range("book", 0, -1));//键值集合
//        out(vo.rangeWithScores("book", 0, -1));
		out(vo.count("book", 2, 4)); //包含的
		out(vo.rangeByLex("book", RedisZSetCommands.Range.range().lte("f").gte("b")));
		out(vo.rangeByLex("book", RedisZSetCommands.Range.range(), RedisZSetCommands.Limit.limit().count(2).offset(0)));
		out(vo.rangeByLex("book", RedisZSetCommands.Range.range(), RedisZSetCommands.Limit.limit().count(2).offset(2)));
		out(vo.rangeByScore("book", 2, 4));
		out(vo.rangeByScore("book", 2, 4, 0, 2));
//        out(vo.rangeByScoreWithScores("book", 2, 4));
//        out(vo.rangeByScoreWithScores("book", 2, 4, 0, 2));

		out(vo.reverseRange("book", 0, -1));
		out(vo.reverseRangeByScore("book", 2, 4));
		out(vo.reverseRangeByScore("book", 2, 4, 0, 2));

		vo.remove("book","b");
		return "Done";
	}

	public void out(String str){
		System.out.println(str);
	}
	public void out(Collection list){
		System.out.println(list);
	}
	public void out(long num){
		System.out.println(num);
	}
	public void out(boolean num){
		System.out.println(num);
	}
	public void out(DataType num){
		System.out.println(num);
	}
	public void out(Object str){
		System.out.println(str);
	}
	public void out(Set<ZSetOperations.TypedTuple<String>> set){
		for(ZSetOperations.TypedTuple<String> t : set){
			System.out.println(t.getValue()+"---"+t.getScore());
		}
	}

//	localhost:8080/redis/logError?clazzName=a.b.c.SomeError
	@GetMapping("/logError")
	@ResponseBody
	public void logError(String clazzName){
		String zsetName = "ABC"+"-ERROR-COUNTER";
		long endTimestamp = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
		long startTimestamp = LocalDateTime.now().plusMinutes(this.minFrequency * -1).toInstant(ZoneOffset.of("+8")).toEpochMilli();

		this.insertRec(zsetName, clazzName, endTimestamp);
		this.sendMessage(zsetName, clazzName, endTimestamp, startTimestamp);
	}

	//	localhost:8080/redis/getErrorByName?clazzName=a.b.c.SomeError
	@GetMapping("/getErrorByName")
	@ResponseBody
	public User getErrorByName(String clazzName){
		List<User> alErrMsgBean = this.getAlErrMsgBean();
		ConcurrentHashMap<String, User> beanMap = alErrMsgBean.stream().collect(Collectors.toMap(
				e -> String.valueOf(e.getUsername()), e -> e, (a, b) -> b, ConcurrentHashMap::new));
		return beanMap.get(clazzName);
	}

	private void insertRec(String zsetName, String clazzName,long endTimestamp){
		ZSetOperations<String, String> opt = this.redisStringTemplate.opsForZSet();
		//成员按值递减(从大到小)来排列
		Set<String> keys = opt.reverseRange(zsetName, 0, -1);
		if(null == keys){ //集合为空则新增
			StringBuffer combineKeyBuf = new StringBuffer();
			combineKeyBuf.append(clazzName).append(":").append(1);
			opt.add(zsetName, combineKeyBuf.toString(),endTimestamp);
		} else {
			boolean hitRec = false;
			for (String key : keys) {
				String[] oldCombineKey = key.split(":");
				if(StringUtils.equals(oldCombineKey[0],clazzName)){
					int count = Integer.valueOf(oldCombineKey[1])+1;
					StringBuffer combineKeyBuf = new StringBuffer();
					combineKeyBuf.append(oldCombineKey[0]).append(":").append(count);
					//键值命中则加1
					opt.add(zsetName, combineKeyBuf.toString(),endTimestamp);
					hitRec = true;
					break;
				}
			}
			//键值空命中则新增
			if(!hitRec){
				StringBuffer combineKeyBuf = new StringBuffer();
				combineKeyBuf.append(clazzName).append(":").append(1);
				opt.add(zsetName, combineKeyBuf.toString(),endTimestamp);
			}
		}
	}

	private void sendMessage(String zsetName, String clazzName,long endTimestamp, long startTimestamp){
		ZSetOperations<String, String> opt = this.redisStringTemplate.opsForZSet();
		//获取时间范围内集合
		Set<String> rangeAscKeys = opt.rangeByScore(zsetName, startTimestamp, endTimestamp);
		Set<String> rangeDescKeys = opt.reverseRangeByScore(zsetName, startTimestamp, endTimestamp);

		rangeAscKeys.forEach(s ->{
			log.info("hit key=>{}",s);
		});
		String rangeStartKey = "";
		String rangeEndKey = "";
		for (String key : rangeAscKeys) {
			String[] oldCombineKey = key.split(":");
			if(StringUtils.equals(oldCombineKey[0],clazzName)){
				rangeStartKey = key;
				break;
			}
		}
		for (String key : rangeDescKeys) {
			String[] oldCombineKey = key.split(":");
			if(StringUtils.equals(oldCombineKey[0],clazzName)){
				rangeEndKey = key;
				break;
			}
		}
		if(StringUtils.isEmpty(rangeStartKey) || StringUtils.isEmpty(rangeEndKey)){
			return;
		}
		String startCount  = rangeStartKey.split(":")[1];
		String endCount  = rangeEndKey.split(":")[1];
		int occurTimes = Integer.valueOf(endCount) - Integer.valueOf(startCount);
		if(occurTimes >= this.countFrequency){
			User errorMessageBean = new User();
			errorMessageBean.setUsername(clazzName);
			errorMessageBean.setPassword("12345");
			errorMessageBean.setRole(1);
			this.addToDetail(errorMessageBean);
			log.info("The error hit {} times and will fire message", occurTimes);
		}
	}

	private void addToDetail(User errorMessageBean){
		String errorBeansKey =  "ABC-ERROR-DETAIL";
		boolean existingRec = false;
		List<Object> messageObjects = redisTemplate.opsForList().range(errorBeansKey, 0, -1);
		if (messageObjects != null && !messageObjects.isEmpty()) {
			for (Object object : messageObjects) {
				User bean = (User) object;
				if(StringUtils.equals(bean.getUsername(),errorMessageBean.getUsername())){
					existingRec = true;
					break;
				}
			}
		}
		if(!existingRec){
			redisTemplate.opsForList().rightPush(errorBeansKey, errorMessageBean);
		}
	}

	public List<User> getAlErrMsgBean() {
		String errorBeansKey = "ABC-ERROR-DETAIL";
		List<Object> messageObjects = redisTemplate.opsForList().range(errorBeansKey, 0, -1);
		List<User> messageBeans = new ArrayList<>();
		if (messageObjects != null && !messageObjects.isEmpty()) {
			for (Object object : messageObjects) {
				messageBeans.add((User) object);
			}
		}
		return messageBeans;
	}
}

